class Subscription:
    def __init__(self, stub, pubsub_name, topic, metadata=None, dead_letter_topic=None):
        # ... existing init code ...
        self._iterator_active = True

    async def __aiter__(self):
        """Make the subscription async iterable."""
        return self

    async def __anext__(self):
        """Get the next message from the subscription stream."""
        if not self._iterator_active:
            raise StopAsyncIteration

        try:
            message = await self.next_message()
            if message is None:
                # If no message is received, continue iteration
                return await self.__anext__()
            return message
        except StreamInactiveError:
            self._iterator_active = False
            raise StopAsyncIteration
        except StreamCancelledError:
            self._iterator_active = False
            raise StopAsyncIteration

    async def close(self):
        """Close the subscription and stop iteration."""
        self._iterator_active = False
        if self._stream:
            try:
                self._stream.cancel()
                self._stream_active.clear()
            except AioRpcError as e:
                if e.code() != StatusCode.CANCELLED:
                    raise Exception(f'Error while closing stream: {e}')
            except Exception as e:
                raise Exception(f'Error while closing stream: {e}')

    # ... other methods stay unchanged ...

async def subscribe_with_handler(
    self,
    pubsub_name: str,
    topic: str,
    handler_fn: Callable[..., Awaitable[TopicEventResponse]],
    metadata: Optional[dict] = None,
    dead_letter_topic: Optional[str] = None,
) -> Callable[[], Awaitable[None]]:
    """
    Subscribe to a topic with a bidirectional stream and a message handler function
    Args:
        pubsub_name (str): The name of the pubsub component.
        topic (str): The name of the topic.
        handler_fn (Callable[..., Awaitable[TopicEventResponse]]): The function to call when a message is received.
        metadata (Optional[dict]): Additional metadata for the subscription.
        dead_letter_topic (Optional[str]): Name of the dead-letter topic.
    Returns:
        Callable[[], Awaitable[None]]: An async function to close the subscription.
    """
    subscription = await self.subscribe(pubsub_name, topic, metadata, dead_letter_topic)

    async def stream_messages():
        try:
            async for message in subscription:
                try:
                    response = await handler_fn(message)
                    if response:
                        await subscription.respond(message, response.status)
                except Exception as e:
                    # Log error and continue processing messages
                    print(f"Error processing message: {e}")
                    continue
        except Exception as e:
            print(f"Stream error: {e}")

    # Start processing messages in background task
    task = asyncio.create_task(stream_messages())

    async def close_subscription():
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass
        await subscription.close()

    return close_subscription

# examples/pubsub-streaming-async/subscriber-iterator.py
import argparse
import asyncio
from dapr.aio.clients import DaprClient
from dapr.clients.grpc._response import TopicEventResponse

parser = argparse.ArgumentParser()
parser.add_argument('--topic', required=True)
args = parser.parse_args()

async def main():
    async with DaprClient() as client:
        subscription = await client.subscribe(
            pubsub_name='pubsub',
            topic=args.topic,
            dead_letter_topic=f'{args.topic}_DEAD'
        )

        try:
            # Process messages using async for
            async for message in subscription:
                print(f'Processing message: {message.data()} from {message.topic()}...')
                # Process message here
                await subscription.respond_success(message)
        except Exception as e:
            print(f"Error: {e}")
        finally:
            await subscription.close()

if __name__ == '__main__':
    asyncio.run(main())

# tests/clients/test_dapr_grpc_client_async.py

async def test_subscribe_topic_iterator(self):
    dapr = DaprGrpcClientAsync(f'{self.scheme}localhost:{self.grpc_port}')
    subscription = await dapr.subscribe(pubsub_name='pubsub', topic='example')
    
    messages = []
    async for message in subscription:
        messages.append(message)
        await subscription.respond_success(message)
        if len(messages) >= 2:  # Get first two messages
            break
            
    self.assertEqual(2, len(messages))
    self.assertEqual('111', messages[0].id())
    self.assertEqual('222', messages[1].id())
    
    await subscription.close()